#include <stdio.h>
#include <stdlib.h>
#include "ProdCons.h"
#include <pthread.h>

void sqsCreate(unsigned int bufferlength) {
    register SQS_PRODCONS *prodcons;

    // Try to allocate object space.
    // We need this to free the gcg.h from using pthread
    if (bufferlength > 0) prodcons = (SQS_PRODCONS*) malloc(sizeof (SQS_PRODCONS));
    else prodcons = NULL;
    if ((this->handle = prodcons) == NULL) return;

    // Ok, we can construct a valid object
    prodcons->buffer = (void**) calloc(bufferlength, sizeof (void*));
    prodcons->buf_size = bufferlength;
    prodcons->mutex_counter = PTHREAD_MUTEX_INITIALIZER;
    prodcons->mutex_cond_prod = PTHREAD_MUTEX_INITIALIZER;
    prodcons->mutex_cond_cons = PTHREAD_MUTEX_INITIALIZER;

    prodcons->prod_cond = PTHREAD_COND_INITIALIZER;
    prodcons->cons_cond = PTHREAD_COND_INITIALIZER;
    prodcons->counter = 0; // The buffer starts with all positions free
    prodcons->prod_pos = 0;
    prodcons->cons_pos = 0;
    prodcons->execute = TRUE;
    setBlocking(true, true);
}

// Creates a buffer with different size

bool createBuffer(unsigned int bufferlength) {
    register SQS_PRODCONS *prodcons = NULL;

    // Check parameter
    destroyBuffer();
    if (bufferlength == 0) return false;

    // Try to allocate object space.
    // We need this to free the gcg.h from using pthread
    prodcons = (SQS_PRODCONS*) malloc(sizeof (SQS_PRODCONS));
    if ((this->handle = prodcons) == NULL) return false;

    // Ok, we can construct a valid object
    prodcons->buffer = (void**) calloc(bufferlength, sizeof (void*));
    prodcons->buf_size = bufferlength;
    prodcons->mutex_counter = PTHREAD_MUTEX_INITIALIZER;
    prodcons->mutex_cond_prod = PTHREAD_MUTEX_INITIALIZER;
    prodcons->mutex_cond_cons = PTHREAD_MUTEX_INITIALIZER;

    prodcons->prod_cond = PTHREAD_COND_INITIALIZER;
    prodcons->cons_cond = PTHREAD_COND_INITIALIZER;
    prodcons->counter = 0; // The buffer starts with all positions free
    prodcons->prod_pos = 0;
    prodcons->cons_pos = 0;
    prodcons->execute = TRUE;
    return true;
}

// Frees all object resources

void destroyBuffer() {
    register SQS_PRODCONS *prodcons = (SQS_PRODCONS*) this->handle;

    if (prodcons != NULL) {
        prodcons->execute = FALSE;
        pthread_cond_broadcast(&prodcons->cons_cond);
        pthread_cond_broadcast(&prodcons->prod_cond);
        if (prodcons->buffer != NULL) free(prodcons->buffer);
        free(prodcons);
    }

    this->handle = NULL;
}

// Destructor

sqsPRODUCERCONSUMER::~sqsPRODUCERCONSUMER() {
    destroyBuffer();
}

// Producer interface

bool put(void *val) {
    register SQS_PRODCONS *prodcons = (SQS_PRODCONS*) this->handle;
    bool status = FALSE;

    // Wait for a free position
    pthread_mutex_lock(&prodcons->mutex_cond_prod);
    while (prodcons->counter == prodcons->buf_size && prodcons->execute && prodcons->blockprod)
        pthread_cond_wait(&prodcons->prod_cond, &prodcons->mutex_cond_prod);

    if (prodcons->execute) {
        // Producers and consumers are racing for this counter
        pthread_mutex_lock(&prodcons->mutex_counter);
        if (prodcons->counter < prodcons->buf_size) { // non-blocking condition
            prodcons->buffer[prodcons->prod_pos] = val;
            prodcons->prod_pos = (prodcons->prod_pos + 1) % prodcons->buf_size;
            status = TRUE;
            prodcons->counter++;
        }
        pthread_mutex_unlock(&prodcons->mutex_counter);
    }
    pthread_mutex_unlock(&prodcons->mutex_cond_prod);


    // Let the consumer get another position
    if (prodcons->execute) pthread_cond_signal(&prodcons->cons_cond);

    return status;
}

// Consumer interface

void * get() {
    register SQS_PRODCONS *prodcons = (SQS_PRODCONS*) this->handle;
    void *val = NULL;

    // Wait for someting to be produced
    pthread_mutex_lock(&prodcons->mutex_cond_cons);
    while (prodcons->counter == 0 && prodcons->execute && prodcons->blockcons)
        pthread_cond_wait(&prodcons->cons_cond, &prodcons->mutex_cond_cons);

    if (prodcons->execute) {
        // Producers and consumers are racing for this counter
        pthread_mutex_lock(&prodcons->mutex_counter);
        if (prodcons->counter > 0) { // Non-blocking test
            val = prodcons->buffer[prodcons->cons_pos];
            prodcons->buffer[prodcons->cons_pos] = NULL;
            prodcons->cons_pos = (prodcons->cons_pos + 1) % prodcons->buf_size;
            prodcons->counter--;
        }
        pthread_mutex_unlock(&prodcons->mutex_counter);
    }
    pthread_mutex_unlock(&prodcons->mutex_cond_cons);

    // Free one buffer's position
    if (prodcons->execute) pthread_cond_signal(&prodcons->prod_cond);

    return val;
}

// Gets the current number of elements

unsigned int getCounter() {
    unsigned int count = 0;
    if (this->handle != NULL) {
        register SQS_PRODCONS *prodcons = (SQS_PRODCONS*) this->handle;
        // Avoid racing conditions
        pthread_mutex_lock(&prodcons->mutex_counter);
        count = prodcons->counter;
        pthread_mutex_unlock(&prodcons->mutex_counter);
    }
    return count;
}

// Pass true if the corresponding end should be blocked if:
//   - buffer is full (producers)
//   - buffer is empty (consumers);

void setBlocking(bool _blockprod, bool _blockcons) {
    register SQS_PRODCONS *prodcons = (SQS_PRODCONS*) this->handle;
    if (prodcons) {
        prodcons->blockcons = _blockcons;
        prodcons->blockprod = _blockprod;
        pthread_cond_broadcast(&prodcons->cons_cond);
        pthread_cond_broadcast(&prodcons->prod_cond);
    }
}


